package com.fjeld.middleware.activemq;

import com.fjeld.middleware.constants.PropertiesManager;
import com.fjeld.middleware.jaxb.ObjectFactory;
import com.fjeld.middleware.jaxb.V1Message;
import com.fjeld.middleware.jaxb.util.V1MessagePool;
import org.apache.log4j.Logger;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import java.io.Serializable;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by IntelliJ IDEA.
 * User: fjeld
 * Date: 02/01/2012
 * Time: 23:49
 *
 * Puts messages into a @BlockingQueue
 */
public class QueuedMessageListener extends TopicListener
{
    private static final Logger LOGGER = Logger.getLogger(QueuedMessageListener.class);
    private BlockingQueue messageQueue;
    private V1MessagePool v1MessagePool = new V1MessagePool();
    private BlockingQueue<V1Message> publisherQueue;

    public QueuedMessageListener(String topicString, String publisherName, BlockingQueue<V1Message> publisherQueue)
    {
        super(topicString, publisherName);
        this.publisherQueue = publisherQueue;
        LOGGER.info(publisherName);

        String queueSizeString = PropertiesManager.getProperties().getProperty(
                "QueuedMessageListener.QueueSize", "1000");
        Integer queueSize = Integer.parseInt(queueSizeString);
        LOGGER.info("QueueSize: " + queueSize);

        this.messageQueue = new LinkedBlockingQueue(queueSize);
    }

    public BlockingQueue getMessageQueue()
    {
        return messageQueue;
    }

    public void startConnection()
    {
        super.startConnection();
    }

    public void onMessage(Message message)
    {
        ObjectMessage objMsg = (ObjectMessage) message;

        try
        {
            Serializable msg = objMsg.getObject();

            if (msg instanceof V1Message)
            {
                // Publish ack
                V1Message v1Message = (V1Message) msg;
                V1Message ack = ObjectFactory.OBJECT_FACTORY.createV1Message();
                ack.setUserID(v1Message.getUserID());
                ack.setCorrespondenceType(v1Message.getCorrespondenceType());
                ack.setCorrespondenceStatus(V1Message.CorrespondenceStatus.Acknowledged.toString());
                ack.setCorrespondenceID(v1Message.getCorrespondenceID());
                ack.setSessionID(v1Message.getSessionID());
                ack.setReceiveTime(System.currentTimeMillis());
                ack.setSendingTime(System.currentTimeMillis());
                publisherQueue.put(ack);

                messageQueue.put(msg);
            } else
            {
                LOGGER.error("Received unsolicited message: " + msg.toString());
            }
        } catch (JMSException e)
        {
            LOGGER.error(logPrefix + "Could not unmarshal message:", e);
        } catch (InterruptedException e)
        {
            LOGGER.error(logPrefix + "Could not put message on message queue:", e);
        }
    }
}